package com.bw.gmall.realtime.Week2Sec.service;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 扩展指标服务类
 * 用于演示如何添加新的统计指标
 */
public class ExtendedMetricsService {
    
    private StreamTableEnvironment tableEnv;
    
    public ExtendedMetricsService(StreamTableEnvironment tableEnv) {
        this.tableEnv = tableEnv;
    }
    
    /**
     * 计算扩展统计指标（示例）
     * @param sourceView 源视图名
     * @return 包含扩展统计结果的表
     */
    public Table calculateExtendedMetrics(String sourceView) {
        String sql = "SELECT " +
                "  window_start, " +
                "  window_end as date_id, " +
                "  shop as shop_id, " +
                // 示例扩展指标
                "  COUNT(*) AS visit_count, " +
                "  AVG(total_amount) AS avg_amount, " +
                "  MAX(total_amount) AS max_amount " +
                "FROM TABLE(TUMBLE(TABLE " + sourceView + ", DESCRIPTOR(rt), INTERVAL '10' MINUTES)) " +
                "GROUP BY window_start, window_end, shop";
        
        return tableEnv.sqlQuery(sql);
    }
    
    /**
     * 创建新的统计表（示例）
     * @param tableName 表名
     */
    public void createNewStatsTable(String tableName) {
        String sql = "CREATE TABLE " + tableName + " (" +
                "  window_start TIMESTAMP(3), " +
                "  date_id STRING, " +
                "  shop_id STRING, " +
                "  visit_count BIGINT, " +
                "  avg_amount DOUBLE, " +
                "  max_amount DOUBLE " +
                ")";
        
        tableEnv.executeSql(sql);
    }
}